// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: Support Cache fast application and slow release mechanism
// feature: Split history as an independent common module
// feature: discovery support client and server
// feature: add statictic lookup
// feature: Crop code for in-process communication
// ------------------------------------------------------------------

#ifndef INCLUDE_EDDS_RTPS_WRITER_STATELESSWRITER_H_
#define INCLUDE_EDDS_RTPS_WRITER_STATELESSWRITER_H_

#ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC

#include <edds/rtps/common/Time_t.h>
#include <history/IChangePool.h>
#include <history/IPayloadPool.h>
#include <edds/rtps/writer/ChangeForReader.h>
#include <edds/rtps/writer/ReaderLocator.h>
#include <edds/rtps/writer/RTPSWriter.h>

#include <condition_variable>
#include <list>
#include <memory>
#include <mutex>
#include <atomic>
#include <ertps/utils/collections/ResourceLimitedVector.hpp>
#include <edds/rtps/interfaces/IReaderDataFilter.hpp>

namespace vbs {
namespace common {
class TimedEvent;

}
}  // namespace vbs

namespace evbs {
namespace ertps {
namespace rtps {
using vbs::common::TimedEvent;
/**
 * Class StatelessWriter, specialization of RTPSWriter that manages writers that don't keep state of
 * the matched readers.
 * @ingroup WRITER_MODULE
 */
class StatelessWriter : public RTPSWriter {
    friend class RTPSParticipantImpl;

 protected:
    StatelessWriter(RTPSParticipantImpl* participant, const GUID_t& guid, const WriterAttributes& attributes,
                    edds::rtps::FlowController* flow_controller, WriterHistory* history,
                    WriterListener* listener = nullptr);

    StatelessWriter(RTPSParticipantImpl* impl, const GUID_t& guid, const WriterAttributes& att,
                    const std::shared_ptr<vbs::common::IPayloadPool>& payload_pool,
                    edds::rtps::FlowController* flow_controller, WriterHistory* hist, WriterListener* listen = nullptr);

    StatelessWriter(RTPSParticipantImpl* impl, const GUID_t& guid, const WriterAttributes& att,
                    const std::shared_ptr<vbs::common::IPayloadPool>& payload_pool,
                    const std::shared_ptr<vbs::common::IChangePool>& change_pool,
                    edds::rtps::FlowController* flow_controller, WriterHistory* hist, WriterListener* listen = nullptr);

 public:
    virtual ~StatelessWriter();

    /**
     * Add a specific change to all ReaderLocators.
     * @param change Pointer to the change.
     * @param max_blocking_time
     */
    void unsent_change_added_to_history(CacheChange_t* change,
                                        const std::chrono::time_point<std::chrono::steady_clock>& max_blocking_time,
                                        bool in_history) override;

    /**
     * Indicate the writer that a change has been removed by the history due to some HistoryQos
     * requirement.
     * @param change Pointer to the change that is going to be removed.
     * @return True if removed correctly.
     */
    bool change_removed_by_history(CacheChange_t* change) override;

    /**
     * Add a matched reader.
     * @param data Pointer to the ReaderProxyData object added.
     * @return True if added.
     */
    bool matched_reader_add(ReaderProxyData& data) override;

    /**
     * Remove a matched reader.
     * @param reader_guid GUID of the reader to remove.
     * @return True if removed.
     */
    bool matched_reader_remove(const GUID_t& reader_guid) override;

    /**
     * Tells us if a specific Reader is matched against this writer
     * @param reader_guid GUID of the reader to check.
     * @return True if it was matched.
     */
    bool matched_reader_is_matched(const GUID_t& reader_guid) override;

    /**
     * Update the Attributes of the Writer.
     * @param att New attributes
     */
    void updateAttributes(const WriterAttributes& att) override {  // LCOV_EXCL_START
        (void)att;
        // FOR NOW THERE IS NOTHING TO UPDATE.
    }  // LCOV_EXCL_STOP

    bool set_fixed_locators(const LocatorList_t& locator_list);

    //! Reset the unsent changes.
    void unsent_changes_reset();

    /**
     * @brief Check if a specific change has been delivered to the transport layer at least once for every matched
     * remote RTPSReader.
     *
     * @param seq_num Sequence number of the change to check.
     * @return true if delivered.
     * @return false otherwise.
     */
    bool has_been_fully_delivered(const SequenceNumber_t& seq_num) const override;

    bool is_acked_by_all(const CacheChange_t* change) const override;

    //! Check if a specific sequence number has been sent to every remote RTPSReader
    bool is_acked_by_all(const SequenceNumber_t& seq_num) const;

    bool try_remove_change(const std::chrono::steady_clock::time_point&,
                           std::unique_lock<vbs::common::RecursiveTimedMutex>&) override;

    bool wait_for_acknowledgement(const SequenceNumber_t& seq,
                                  const std::chrono::steady_clock::time_point& max_blocking_time_point,
                                  std::unique_lock<vbs::common::RecursiveTimedMutex>& lock) override;

    /**
     * Send a message through this interface.
     *
     * @param message Pointer to the buffer with the message already serialized.
     * @param locator_selector RTPSMessageSenderInterface reference uses for selecting locators. The
     * reference has to be a member of this RTPSWriter object.
     * @param max_blocking_time_point Future timepoint where blocking send should end.
     */
    bool send_nts(CDRMessage_t* message, const LocatorSelectorSender& locator_selector) const override;

    /**
     * Get the number of matched readers
     * @return Number of the matched readers
     */
    size_t getMatchedReadersSize() const override;

    /*!
     * Tells writer the sample can be sent to the network.
     * This function should be used by a edds::rtps::FlowController.
     *
     * @param cache_change Pointer to the CacheChange_t that represents the sample which can be
     * sent.
     * @param group RTPSMessageGroup reference uses for generating the RTPS message.
     * @param locator_selector RTPSMessageSenderInterface reference uses for selecting locators. The
     * reference has to be a member of this RTPSWriter object.
     * @param max_blocking_time Future timepoint where blocking send should end.
     * @return Return code.
     * @note Must be non-thread safe.
     */
    DeliveryRetCode deliver_sample_nts(CacheChange_t* cache_change, RTPSMessageGroup& group,
                                       LocatorSelectorSender& locator_selector,
                                       const std::chrono::time_point<std::chrono::steady_clock>& max_blocking_time,
                                       bool in_history = true) override;
    LocatorSelectorSender& get_general_locator_selector() override { return locator_selector_; }
    LocatorSelectorSender& get_async_locator_selector() override { return locator_selector_; }

    edds::dds::builtin::StatisticMatchGuids get_remote_guids() override;

    edds::dds::builtin::StatisticProxyInfos get_proxy_infos() override;

    void FilterReaderLocators(ReaderProxyData& rdata, ReaderLocator* reader);

 private:
    void init(RTPSParticipantImpl* participant, const WriterAttributes& attributes);

    void get_builtin_guid();

    bool has_builtin_guid();

    void update_reader_info(bool create_sender_resources);

    bool intraprocess_delivery(CacheChange_t* change, ReaderLocator& reader_locator);

    bool history_adjustment();

    LocatorList_t fixed_locators_;
    ResourceLimitedVector<std::unique_ptr<ReaderLocator>> matched_remote_readers_;

    std::condition_variable_any unsent_changes_cond_;

    uint64_t current_sequence_number_sent_ = 0U;

    FragmentNumber_t current_fragment_sent_ = 0U;

    std::atomic<uint64_t> last_sequence_number_sent_ {0U};

    ResourceLimitedVector<std::unique_ptr<ReaderLocator>> matched_local_readers_;

    ResourceLimitedVector<std::unique_ptr<ReaderLocator>> matched_readers_pool_;

    LocatorSelectorSender locator_selector_;

    vbs::common::TimedEvent* free_event_ = nullptr;

    uint32_t mem_watermark = 0;
};

} /* namespace rtps */
} /* namespace ertps */
} /* namespace evbs */

#endif  // ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC
#endif  // INCLUDE_EDDS_RTPS_WRITER_STATELESSWRITER_H_
